package com.ydlclass.proxy.handler;

import com.ydlclass.NettyBootStrapInitializer;
import com.ydlclass.YrpcBootstrap;
import com.ydlclass.compress.Factory.CompressFactory;
import com.ydlclass.discovery.Registry;
import com.ydlclass.enumeration.RequestType;
import com.ydlclass.exceptions.DiscoveryException;
import com.ydlclass.exceptions.NetworkException;
import com.ydlclass.serialize.Factory.SerializeFactory;
import com.ydlclass.transport.message.RequestPayload;
import com.ydlclass.transport.message.YrpcRequest;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.util.Date;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 该类封装了客户端通信的基础逻辑，每一个代理对象的远程调用过程都封装在invoke方法中
 * 1、发现可用服务
 * 2、建立连接
 * 3、发送请求
 * 4、得到结果
 * @author Knife
 */
@Slf4j
public class RpcConsumerInvocationHandler implements InvocationHandler {
    // 需要一个注册中心 和一个接口
    private final Registry registry;

    private final Class<?> interfaceRef;

    public RpcConsumerInvocationHandler(Registry registry, Class<?> interfaceRef) {
        this.registry = registry;
        this.interfaceRef = interfaceRef;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        // 我们调用sayHi 方法，事实上会走进这个代码段中
        // 我们已经知道method(具体的方法)。args
        // log.debug("服务注册进行中");
        // todo q: 我们每次调用相关方法的时候都需要去注册中心拉去服务列表吗？  本地缓存+Watcher(监视)
        // todo    我们如何合理的选择一个可用的服务，而不是只获取第一个    负载均衡


        /**
         * --------------------------封装报文------------------------------
         * 建造者模式
         */
        // 1、 封装报文
        RequestPayload payload = RequestPayload.builder()
                .interfaceName(interfaceRef.getName())
                .methodName(method.getName())
                .parametersType(method.getParameterTypes())
                .parametersValue(args)
                .returnType(method.getReturnType())
                .build();

        // todo 需要对请求id和各种类型做处理

        YrpcRequest yrpcRequest = YrpcRequest.builder()
                .requestId(YrpcBootstrap.ID_GENERATOR.getId())
                .compressType(CompressFactory.getCompressor(YrpcBootstrap.COMPRESSOR_TYPE).getCode())
                .requestType(RequestType.REQUEST.getId())
                .serializeType(SerializeFactory.getSerializer(YrpcBootstrap.SERIALIZE_TYPE).getCode())
                .timeStamp(System.currentTimeMillis())
                .requestPayload(payload)
                .build();

        // 将请求存入线程本地，需要在合适的时候调用remove
        YrpcBootstrap.THREAD_LOCAL.set(yrpcRequest);




        // 2、尝试获取当前配置的负载均衡器，获取可用节点，从服务端拉去列表，利用负载均衡策略，进行负载均衡选取可用节点；
        InetSocketAddress address = YrpcBootstrap
                .LOAD_BALANCE_STRATEGY
                .selectServiceAddress(interfaceRef.getName());

        if (log.isDebugEnabled()){
            log.debug("服务调用方发现了服务{}的可用节点{}",interfaceRef.getName(),address);
        }

        // 使用netty连接服务器，发送调用的服务的名字+ 方法名称 + 参数列表  ，得到结果
        // todo ：意味着每次调用都会产生一个netty
        // todo  也就意味着每次在这里新加一个连接是不可以的
        // 解决方案 ： 缓存channel  尝试从缓存中获取已经连接的channel，如果获取不到则创建新的连接，并进行缓存

        // 2、 尝试从全局缓存中获取一个可用的通道

        Channel channel = getAvailableChannel(address);


        /**
         * ---------------------同步策略-------------------------------
         *                 ChannelFuture channelFuture = channel.writeAndFlush(new Object());
         *                 // 需要学习channelFuture 的简单api
         *                 if (channelFuture.isDone()){
         *                     Object object = channelFuture.getNow();
         *                 } else if (!channelFuture.isSuccess()){
         *                     需要捕获异常，可以捕获异步任务中的异常
         *                     Throwable cause = channelFuture.cause();
         *                     throw new NetworkException(cause);
         *                 }
         */

        /**
         * ---------------------异步策略-------------------------------
         */


        CompletableFuture<Object> completableFuture = new CompletableFuture<>();
        // 需要将completableFuture暴露出去

        YrpcBootstrap.PENDING_REQUEST.put(yrpcRequest.getRequestId(),completableFuture);

        // 这里封装的请求实例会writeAndFlush写出一个请求，这个请求实例就会进入pipeline执行出站的一系列操作之后
        // 我们就可以，第一个出站的程序就会将yrpcRequest --> 二进制的报文
        channel.writeAndFlush(yrpcRequest).addListener((ChannelFutureListener) promise -> {
            /**
             * 将来的promise将来返回的结果是什么？ writeAndFlush的返回结果
             * 一旦数据被写出去，promise也就结束了
             * 我们得到结果是什么？服务器端给我们返回的结果，所以我们处理的completableFuture可能会有问题
             * 服务端返回的结果是异步的，应该将completableFuture挂起并且暴露，再得到服务提供方响应的时候调用complete方法
             */

            if (!promise.isSuccess()){
                // 获取异常
                completableFuture.completeExceptionally(promise.cause());
            }
        });
        // 如果没有地方处理completableFuture，这里会阻塞,等待complete方法的执行
        // q:我们需要在哪里调用complete来进行得到结果，很明显 pipeline 中最终的handler的处理结果
        // 获得响应的结果

        // 清理ThreadLocal里面的数据，避免内存泄漏
        YrpcBootstrap.THREAD_LOCAL.remove();


        // todo 会等待complete执行完毕
        return completableFuture.get(10, TimeUnit.SECONDS);


    }

    /**
     *  根据地址获取一个可用的通道
     * @param address
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     * @throws TimeoutException
     */
    private Channel getAvailableChannel(InetSocketAddress address)  {

        // 1、尝试从缓存中获取
        Channel channel = YrpcBootstrap.CHANNEL_CACHE.get(address);

        // 2、拿不到就去建立连接
        if (channel == null ){
            // 建立一个新的连接并且放入缓存中

            // await 会阻塞线程，等待连接成功之后在进行返回 ；；netty同样也提供一步处理的addListener
            // sync 和await 都是阻塞当前线程，获取返回值(连接的过程是异步的，发送数据的过程是异步的)
            // sync 如果发生了异常sync会主动在主线程抛出异常，await不会，await是开辟子线程，异常在子线程中处理

            // 同步方式获得channel
            //  channel = NettyBootStrapInitializer
            //       .getBootstrap()
            //       .connect(address)
            //       .await()
            //       .channel();

            // 异步方式获得channel

            CompletableFuture<Channel> channelFuture = new CompletableFuture<>();

            // addListener这是一个线程，异步的方式进行，要想获取这个线程里面的结果，需要CompletableFuture
            NettyBootStrapInitializer.getBootstrap().connect(address).addListener((ChannelFutureListener) promise -> {
                if (promise.isDone()){
                    // 异步的，我们已经完成
                    if (log.isDebugEnabled()){
                        log.debug("已经和{}建立连接", address);
                    }
                    channelFuture.complete(promise.channel());
                } else if (!promise.isSuccess()){
                    channelFuture.completeExceptionally(promise.cause());
                }
            });

            // 阻塞获取channel
            try {
                channel = channelFuture.get(3, TimeUnit.SECONDS);
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                log.error("获取通道时发生异常");
                throw new DiscoveryException(e);
            }

            // 缓存
            YrpcBootstrap.CHANNEL_CACHE.put(address,channel);

        }


        if (channel == null ){
            log.debug("获取{}通道发生异常",address);
            throw new NetworkException("获取channel通道发生异常");
        }

        return channel;
    }
}
